package com.lichao.FinkDataProcess.etl;

/**
 * @author: li chao
 * @Desc:
 * @create: 2024-11-20 12:00
 **/
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.util.Collector;

import java.util.Properties;
import java.util.Random;

public class FlinkKafkaRandomPrefix {

    public static void main(String[] args) throws Exception {
        // 设置执行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 配置 Kafka 消费者属性
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");
        properties.setProperty("group.id", "flink-consumer-group");

        // 创建 Kafka 消费者
        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>(
                "input-topic", // 输入主题
                new SimpleStringSchema(), // 反序列化器
                properties
        );

        // 从 Kafka 消费数据
        DataStream<String> stream = env.addSource(consumer);

        // 添加随机前缀并发送到另一个 Kafka 主题
        DataStream<String> processedStream = stream.flatMap(new RandomPrefixMapper());

        // 配置 Kafka 生产者属性
        Properties producerProps = new Properties();
        producerProps.setProperty("bootstrap.servers", "localhost:9092");

        // 创建 Kafka 生产者
        FlinkKafkaProducer<String> producer = new FlinkKafkaProducer<>(
                "output-topic", // 输出主题
                new SimpleStringSchema(), // 序列化器
                producerProps
        );

        // 将处理后的数据发送到 Kafka
        processedStream.addSink(producer);

        // 启动执行环境
        env.execute("Flink Kafka Random Prefix Example");
    }

    // 自定义 FlatMapFunction 来添加随机前缀
    public static class RandomPrefixMapper implements org.apache.flink.api.common.functions.FlatMapFunction<String, String> {
        private static final Random RANDOM = new Random();
        private static final String CHARACTERS = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789";
        private static final int PREFIX_LENGTH = 5; // 随机前缀的长度

        @Override
        public void flatMap(String value, Collector<String> out) {
            StringBuilder prefix = new StringBuilder(PREFIX_LENGTH);
            for (int i = 0; i < PREFIX_LENGTH; i++) {
                prefix.append(CHARACTERS.charAt(RANDOM.nextInt(CHARACTERS.length())));
            }
            out.collect(prefix.toString() + value);
        }
    }
}